package net.bwie.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.SneakyThrows;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;

/**
 * 从Kafka消息队列实时消费采集交易订单数据，基于窗口window进行聚合计算
 * @author xuanyu
 * @date 2025/10/27
 */
public class _05RealtimeWindowOrder {

	public static void main(String[] args) throws Exception {
		// 1. 执行环境-env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setParallelism(1) ;

		// 2. 数据源-source
		KafkaSource<String> source = KafkaSource.<String>builder()
			.setBootstrapServers("node101:9092,node102:9092,node103:9092")
			.setTopics("foo-order")
			.setGroupId("realtime-g1")
			.setStartingOffsets(OffsetsInitializer.earliest())
			.setValueOnlyDeserializer(new SimpleStringSchema())
			.build();
		DataStreamSource<String> stream = env.fromSource(
			source, WatermarkStrategy.noWatermarks(), "Kafka Source"
		);

		// todo 对数据进行处理，并且保存结果
		// todo 需求1:
		//  使用Flink  API读取Kafka中的数据，开启10分钟窗口，统计每个区域（areaId）所产生的的订单总数量。（要有窗口开始时间，结束时间，订单总数）
		computeArea(stream) ;

		// todo 需求2:
		//  使用Flink  API，开启10分窗口，每2秒滑动一次，统计已退款的订单信息（要有窗口开始时间，结束时间，已退款信息） 。
		computeRefund(stream) ;

		// todo 需求3:
		//  使用Flink  API开启10分钟滚动窗口，统计每个用户所实际支付订单的总金额。
		computePay(stream) ;

		// todo 需求4:
		//  使用Flink API开启10分钟窗口，统计每个区域实际支付订单最高的TOP3用户信息。（要有窗口开始时间，结束时间，TOP3用户信息）。
		computeAreaTop(stream);

		// 5. 触发执行
		env.execute("RealtimeWindowOrder") ;
	}

	/**
	 * 使用Flink API开启10分钟窗口，统计每个区域实际支付订单最高的TOP3用户信息。（要有窗口开始时间，结束时间，TOP3用户信息）。
	 */
	private static void computeAreaTop(DataStreamSource<String> stream) {
		// s1.指定数据时间
		SingleOutputStreamOperator<String> stream1 = stream.assignTimestampsAndWatermarks(
			WatermarkStrategy
				.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
				.withTimestampAssigner(new SerializableTimestampAssigner<String>() {
					SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") ;
					@SneakyThrows
					@Override
					public long extractTimestamp(String element, long recordTimestamp) {
						// 外层解析json
						JSONObject jsonObject = JSON.parseObject(element);
						// 获取after字段值
						JSONObject afterJsonObject = jsonObject.getJSONObject("after");
						// 获取订单时间
						String createTimeValue = afterJsonObject.getString("createTime");
						// 转换
						Date date = format.parse(createTimeValue);
						// 时间戳
						return date.getTime();
					}
				})
		);

		// s2.获取字段值
		SingleOutputStreamOperator<Tuple2<String, Double>> stream2 = stream1.map(
			new MapFunction<String, Tuple2<String, Double>>() {
				@Override
				public Tuple2<String, Double> map(String value) throws Exception {
					// 外层解析json
					JSONObject jsonObject = JSON.parseObject(value);
					// 获取after字段值
					JSONObject afterJsonObject = jsonObject.getJSONObject("after");
					// areaId 字段值
					String areaIdValue = afterJsonObject.getString("areaId");
					// areaId 字段值
					Double realTotalMoneyValue = afterJsonObject.getDouble("realTotalMoney");
					// 返回二元组
					return Tuple2.of(areaIdValue, realTotalMoneyValue);
				}
			}
		);

		// s3. 分组
		KeyedStream<Tuple2<String, Double>, String> stream3 = stream2.keyBy(tuple -> tuple.f0);

		// s4. 设置窗口
		WindowedStream<Tuple2<String, Double>, String, TimeWindow> stream4 = stream3.window(
			TumblingEventTimeWindows.of(Time.minutes(10))
		);

		// s5. 窗口计算，求和
		SingleOutputStreamOperator<String> stream5 = stream4.apply(
			new WindowFunction<Tuple2<String, Double>, String, String, TimeWindow>() {
				private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") ;
				@Override
				public void apply(String areaId,
				                  TimeWindow window,
				                  Iterable<Tuple2<String, Double>> input,
				                  Collector<String> out) throws Exception {
					// 窗口开始和结束时间
					String windowStart = format.format(window.getStart());
					String windowEnd = format.format(window.getEnd());
					// 金额求和
					double sum = 0.0 ;
					for (Tuple2<String, Double> tuple : input) {
						sum += tuple.f1;
					}
					// 输出
					String output = windowStart + "," + windowEnd + "," + areaId + "," + sum ;
					out.collect(output);
				}
			}
		);

		//stream5.print("area-pay");

		/*
-- 创建数据库
CREATE DATABASE IF NOT EXISTS db_flink ;
-- 创建表
CREATE TABLE IF NOT EXISTS db_flink.tbl_area_pay_report(
    window_start_time VARCHAR(255),
    window_end_time VARCHAR(255),
    area_id VARCHAR(255),
    pay_amount DOUBLE,
    PRIMARY KEY (window_start_time, window_end_time, area_id)
);

REPLACE INTO db_flink.tbl_area_pay_report(window_start_time, window_end_time, area_id, pay_amount) VALUES (?, ?, ?, ?);
		 */
		SinkFunction<String> jdbcSink = JdbcSink.sink(
			"REPLACE INTO db_flink.tbl_area_pay_report(window_start_time, window_end_time, area_id, pay_amount) VALUES (?, ?, ?, ?) ",
			(statement, value) -> {
				// value -> 2022-08-31 20:30:00,2022-08-31 20:40:00,37,753.0
				String[] split = value.split(",");
				// 设置值
				statement.setObject(1, split[0]);
				statement.setObject(2, split[1]);
				statement.setObject(3, split[2]);
				statement.setObject(4, split[3]);
			},
			JdbcExecutionOptions.builder()
				.withBatchSize(1000)
				.withBatchIntervalMs(200)
				.withMaxRetries(5)
				.build(),
			new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
				.withUrl("jdbc:mysql://node101:3306/db_flink")
				.withDriverName("com.mysql.cj.jdbc.Driver")
				.withUsername("root")
				.withPassword("123456")
				.build()
		);
		stream5.addSink(jdbcSink);

	}

	/**
	 * 使用Flink  API开启10分钟滚动窗口，统计每个用户所实际支付订单的总金额。
	 */
	private static void computePay(DataStreamSource<String> stream) {
		// s1.指定数据时间
		SingleOutputStreamOperator<String> stream1 = stream.assignTimestampsAndWatermarks(
			WatermarkStrategy
				.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
				.withTimestampAssigner(new SerializableTimestampAssigner<String>() {
					SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") ;
					@SneakyThrows
					@Override
					public long extractTimestamp(String element, long recordTimestamp) {
						// 外层解析json
						JSONObject jsonObject = JSON.parseObject(element);
						// 获取after字段值
						JSONObject afterJsonObject = jsonObject.getJSONObject("after");
						// 获取订单时间
						String createTimeValue = afterJsonObject.getString("createTime");
						// 转换
						Date date = format.parse(createTimeValue);
						// 时间戳
						return date.getTime();
					}
				})
		);

		// s2.获取字段值
		SingleOutputStreamOperator<Tuple2<String, Double>> stream2 = stream1.map(
			new MapFunction<String, Tuple2<String, Double>>() {
				@Override
				public Tuple2<String, Double> map(String value) throws Exception {
					// 外层解析json
					JSONObject jsonObject = JSON.parseObject(value);
					// 获取after字段值
					JSONObject afterJsonObject = jsonObject.getJSONObject("after");
					// userId 字段值
					String userIdValue = afterJsonObject.getString("userId");
					// areaId 字段值
					Double realTotalMoneyValue = afterJsonObject.getDouble("realTotalMoney");
					// 返回二元组
					return Tuple2.of(userIdValue, realTotalMoneyValue);
				}
			}
		);

		// s3. 分组
		KeyedStream<Tuple2<String, Double>, String> stream3 = stream2.keyBy(tuple -> tuple.f0);

		// s4. 设置窗口
		WindowedStream<Tuple2<String, Double>, String, TimeWindow> stream4 = stream3.window(
			TumblingEventTimeWindows.of(Time.minutes(10))
		);

		// s5. 窗口计算，求和
		SingleOutputStreamOperator<String> stream5 = stream4.apply(
			new WindowFunction<Tuple2<String, Double>, String, String, TimeWindow>() {
				private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") ;
				@Override
				public void apply(String userId,
				                  TimeWindow window,
				                  Iterable<Tuple2<String, Double>> input,
				                  Collector<String> out) throws Exception {
					// 窗口开始和结束时间
					String windowStart = format.format(window.getStart());
					String windowEnd = format.format(window.getEnd());
					// 金额求和
					double sum = 0.0 ;
					for (Tuple2<String, Double> tuple : input) {
						sum += tuple.f1;
					}
					// 输出
					String output = windowStart + "," + windowEnd + "," + userId + "," + sum ;
					out.collect(output);
				}
			}
		);

		// 结果打印控制台
		stream5.print("pay");

	}

	/**
	 * 使用Flink  API，开启10分窗口，每2秒滑动一次，统计已退款的订单信息（要有窗口开始时间，结束时间，已退款信息） 。
	 */
	private static void computeRefund(DataStreamSource<String> stream) {
		// s1.指定数据时间
		SingleOutputStreamOperator<String> stream1 = stream.assignTimestampsAndWatermarks(
			WatermarkStrategy
				.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
				.withTimestampAssigner(new SerializableTimestampAssigner<String>() {
					SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") ;
					@SneakyThrows
					@Override
					public long extractTimestamp(String element, long recordTimestamp) {
						// 外层解析json
						JSONObject jsonObject = JSON.parseObject(element);
						// 获取after字段值
						JSONObject afterJsonObject = jsonObject.getJSONObject("after");
						// 获取订单时间
						String createTimeValue = afterJsonObject.getString("createTime");
						// 转换
						Date date = format.parse(createTimeValue);
						// 时间戳
						return date.getTime();
					}
				})
		);

		// s2.获取退款订单数据
		SingleOutputStreamOperator<String> stream2 = stream1.filter(
			new FilterFunction<String>() {
				@Override
				public boolean filter(String value) throws Exception {
					// 外层解析json
					JSONObject jsonObject = JSON.parseObject(value);
					// 获取after字段值
					JSONObject afterJsonObject = jsonObject.getJSONObject("after");
					// 提取字段值isRefund
					Integer isRefundValue = afterJsonObject.getInteger("isRefund");
					// 比较
					return 1 == isRefundValue;
				}
			}
		);

		// s3.分组
		KeyedStream<String, String> stream3 = stream2.keyBy(json -> "refund");

		// s4. 窗口
		WindowedStream<String, String, TimeWindow> stream4 = stream3.window(
			SlidingEventTimeWindows.of(Time.minutes(10), Time.seconds(2))
		);

		// s5.窗口计算
		SingleOutputStreamOperator<String> stream5 = stream4.apply(
			new WindowFunction<String, String, String, TimeWindow>() {
				private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") ;
				@Override
				public void apply(String string,
				                  TimeWindow window,
				                  Iterable<String> input,
				                  Collector<String> out) throws Exception {
					// 窗口开始和结束时间
					String windowStart = format.format(window.getStart());
					String windowEnd = format.format(window.getEnd());

					// todo 要有窗口开始时间，结束时间，已退款信息
					for (String value : input) {
						out.collect(windowStart + "$" + windowEnd + "$" + value);
					}
				}
			}
		);

		// 结果打印控制台
		stream5.print("refund");
	}

	/**
	 * 使用Flink  API读取Kafka中的数据，开启10分钟窗口，统计每个区域（areaId）所产生的的订单总数量。（要有窗口开始时间，结束时间，订单总数）
	 */
	private static void computeArea(DataStreamSource<String> stream) {
		// s1.指定数据时间
		SingleOutputStreamOperator<String> stream1 = stream.assignTimestampsAndWatermarks(
			WatermarkStrategy
				.<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
				.withTimestampAssigner(new SerializableTimestampAssigner<String>() {
					SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") ;
					@SneakyThrows
					@Override
					public long extractTimestamp(String element, long recordTimestamp) {
						// 外层解析json
						JSONObject jsonObject = JSON.parseObject(element);
						// 获取after字段值
						JSONObject afterJsonObject = jsonObject.getJSONObject("after");
						// 获取订单时间
						String createTimeValue = afterJsonObject.getString("createTime");
						// 转换
						Date date = format.parse(createTimeValue);
						// 时间戳
						return date.getTime();
					}
				})
		);

		// s2.获取字段值
		SingleOutputStreamOperator<Tuple2<String, Long>> stream2 = stream1.map(
			new MapFunction<String, Tuple2<String, Long>>() {
				@Override
				public Tuple2<String, Long> map(String value) throws Exception {
					// 外层解析json
					JSONObject jsonObject = JSON.parseObject(value);
					// 获取after字段值
					JSONObject afterJsonObject = jsonObject.getJSONObject("after");
					// areaId 字段值
					String areaIdValue = afterJsonObject.getString("areaId");
					// 返回二元组
					return Tuple2.of(areaIdValue, 1L);
				}
			}
		);

		// s3.区域分组
		KeyedStream<Tuple2<String, Long>, String> stream3 = stream2.keyBy(tuple -> tuple.f0);

		// s4.设置窗口：10分钟窗口
		WindowedStream<Tuple2<String, Long>, String, TimeWindow> stream4 = stream3.window(
			TumblingEventTimeWindows.of(Time.minutes(10))
		);

		// s5.窗口计算：订单总数量
		SingleOutputStreamOperator<String> stream5 = stream4.apply(
			new WindowFunction<Tuple2<String, Long>, String, String, TimeWindow>() {
				private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") ;
				@Override
				public void apply(String areaId,
				                  TimeWindow window,
				                  Iterable<Tuple2<String, Long>> input,
				                  Collector<String> out) throws Exception {
					// 窗口开始和结束时间
					String windowStart = format.format(window.getStart());
					String windowEnd = format.format(window.getEnd());
					// 计算订单数
					long sumOrder = 0 ;
					for (Tuple2<String, Long> tuple : input) {
						sumOrder += tuple.f1 ;
					}
					// todo 要有窗口开始时间，结束时间，订单总数
					String output = windowStart + "," + windowEnd + "," + areaId + "," + sumOrder;
					out.collect(output);
				}
			}
		);

		// 打印控制台
		stream5.print("area");
	}


}
